package com.software.process

import com.software.util.DBTools
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions.expr
import org.apache.spark.sql.{DataFrame, SparkSession}

object RegionalTemperatureProcess {

  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    //1.创建SparkSession对象（启动Spark）
    val sparkSession: SparkSession =
      SparkSession.builder().master("local[*]")
        .appName("RegionalTemperatureProcess")
        .config("spark.testing.memory", "2147480000")
        .getOrCreate()
    /*val sparkSession: SparkSession =
      SparkSession.builder()
        .appName("RegionalTemperatureProcess")
        .getOrCreate()*/

    //2.加载csv文件（数据源）

    val inputWebfile = "input/stateTemperatures.csv"
    var weatherData: DataFrame = sparkSession.read.format("csv")
      .option("header", true)
      .option("multiLine", true)
      .load(inputWebfile)

    //正则表达式的替换
    weatherData = regexpReplace(weatherData, "dt", "/", "-")
    weatherData.show(5)

    //3.注册临时表
    weatherData.createOrReplaceTempView("tbl_globalLandTemperaturesByState")
    //4.统计中国不同省份每个月的气温
    val result: DataFrame =
      sparkSession.sql(
        """
          |SELECT state,substring(dt,0,4) year,
          |sum(CasE month(dt) WHEN 1 THEN ROUND(AverageTemperature,3) ELSE 0 END) as January,
          |sum(CasE month(dt) WHEN 2 THEN ROUND(AverageTemperature,3) ELSE 0 END) as February,
          |sum(CasE month(dt) WHEN 3 THEN ROUND(AverageTemperature,3) ELSE 0 END) as March,
          |sum(CasE month(dt) WHEN 4 THEN ROUND(AverageTemperature,3) ELSE 0 END) as April,
          |sum(CasE month(dt) WHEN 5 THEN ROUND(AverageTemperature,3) ELSE 0 END) as May,
          |sum(CasE month(dt) WHEN 6 THEN ROUND(AverageTemperature,3) ELSE 0 END) as June,
          |sum(CasE month(dt) WHEN 7 THEN ROUND(AverageTemperature,3) ELSE 0 END) as July,
          |sum(CasE month(dt) WHEN 8 THEN ROUND(AverageTemperature,3) ELSE 0 END) as August,
          |sum(CasE month(dt) WHEN 9 THEN ROUND(AverageTemperature,3) ELSE 0 END) as September,
          |sum(CasE month(dt) WHEN 10 THEN ROUND(AverageTemperature,3) ELSE 0 END) as October,
          |sum(CasE month(dt) WHEN 11 THEN ROUND(AverageTemperature,3) ELSE 0 END) as November,
          |sum(CasE month(dt) WHEN 12 THEN ROUND(AverageTemperature,3) ELSE 0 END) as December
          |FROM tbl_globalLandTemperaturesByState
          |where Country='China'
          |group by state,substring(dt,0,4)
          |order by substring(dt,0,4)
          |""".stripMargin)

    result.show(false)
    DBTools.WriteMySql("china_province_temperatures", result)

    //5.统计中国不同地区每年每季度的平均气温
    val chinaresult: DataFrame =
      sparkSession.sql(
        """
          |SELECT state ,YEAR(dt) AS year,round(AVG(averageTemperature),3) AS AverageSeasonTemperature,CEIL(MONTH(dt)/3) AS season
          |FROM  tbl_globalLandTemperaturesByState
          |where AverageTemperature != ''
          |AND country='China'
          |GROUP BY state,year,season
          |ORDER BY year desc
          | """.stripMargin)

    DBTools.WriteMySql("china_season_compare_temperature", chinaresult)

  }

  //正则替换
  def regexpReplace(df: DataFrame, columnName: String, regexp: String, newValue: Any): DataFrame = {
    val exprString: String = "regexp_replace(" + columnName + ",'" + regexp + "','" + newValue + "')"
    df.withColumn(columnName, expr(exprString).alias(columnName))
  }
}
